package io.github.shenbinglife.jlock.hdfs;

import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * JReentrantHadoopFileLock
 *
 * @author shenbing
 * @date 2022/4/21
 */
public class JReentrantHadoopFileLock extends JAsyncHadoopFileLock {

  private static final Logger logger = LoggerFactory.getLogger(JReentrantHadoopFileLock.class);

  private AtomicInteger lockCount = new AtomicInteger(0);
  private Thread lockThread = null;

  public JReentrantHadoopFileLock(Path lockFile) {
    super(lockFile);
  }

  public JReentrantHadoopFileLock(Path lockFile, long timeoutMillis) {
    super(lockFile, timeoutMillis);
  }

  public JReentrantHadoopFileLock(Path lockFile, long timeoutMillis, long refreshPeriodMillis) {
    super(lockFile, timeoutMillis, refreshPeriodMillis);
  }

  @Override
  protected boolean tryLockOnce() {
    if (isLocked() && Thread.currentThread() == lockThread) {
      lockCount.incrementAndGet();
      return true;
    } else {
      boolean success = super.tryLockOnce();
      if (success) {
        lockCount.getAndIncrement();
        lockThread = Thread.currentThread();
      }
      return success;
    }
  }

  @Override
  public void unlock() {
    if (isLocked()) {
      if (Thread.currentThread() == lockThread) {
        if (lockCount.decrementAndGet() == 0) {
          super.unlock();
        }
      }

    } else {
      logger.warn("Lock may expire when unlock, we need care it's safety");
    }
  }
}
